|
Atividade Integradora¶
|
- Big Data e Computação em Nuvem / Modelos Preditivos Avançados
Insper - Jul/2020 - São Paulo, Brasil
- Autores:
- Marcelo Franceschini
- Rafael Costa
- Viviane Sanchez
Introdução¶
Foram selecinados dados de usuários e estabelecimentos abertos em Toronto para treinar um modelo de classificação binária das notas das avaliações e assim prever se a nota de um estabelecimento que o usuário ainda não avaliou seria boa ou ruim. A recomendação é dada conforme a maior probabilidade de o usuário dar uma nota alta para aquele lugar.
Objetivo¶
Criar um sistema de recomendação de estabelecimentos bem avaliados em linha com o perfil do usuário.
Os dados das seguintes bases fornecidas pelp Yelp para estudos acadêmicos serão analisados
- Business - informações e atributos dos estabelecimentos
- Users - Informações sobre o perfil do usuário no Yelp
- Reviews - Texto e atributos de avaliações dos estabelecimentos criadas pelos usuários
- Tips - Dicas adicionais e atributos deixados pelos usuários
Check-ins - Histórico de check-ins por estabelecimento
Ambiente Spark¶
A seguir estão as confirgurações para o ambiente Spark. Como o notebook foi compartilhado por GitHub e processado em máquinas diferentes, foram mantidas as diferentes configurações necessárias.
In [1]:
#import findspark as fs
from pyspark.sql import SparkSession
In [117]:
import matplotlib.pyplot as plt
from wordcloud import WordCloud, ImageColorGenerator
import os
from pyspark.sql import SparkSession
# from pyspark.sql.types import *
from pyspark.sql import functions as f
from pyspark.sql.window import Window
from pyspark.ml.feature import StopWordsRemover
from pyspark.sql.functions import col, count, explode, split, array, concat_ws
from pyspark.sql.types import ArrayType, IntegerType
import pandas as pd
import seaborn as sns
#from tqdm.notebook import tqdm
sns.set(style="ticks", palette="pastel")
%matplotlib inline
In [4]:
#Mac Local (Viviane)
#spark_location='/Users/vivi/server/spark' # Set your own
#java8_location= '/Library/Java/JavaVirtualMachines/jdk1.8.0_251.jdk/Contents/Home/' # Set your own
#os.environ['JAVA_HOME'] = java8_location
#fs.init(spark_home=spark_location)
In [5]:
#datapath = 'C:\\Users\\RuWindows\\Desktop\\PI\\yelp_dataset\\' #Marcelo
datapath = '../data/yelp' #servidor AWS
#datapath = 'data' #Viviane
In [17]:
#files = sorted(os.listdir(datapath))
#files
In [18]:
#!head data/yelp_academic_dataset_review.json
In [6]:
# Spark Session
spark = SparkSession.builder \
.master('local[*]') \
.appName('Integradora Yelp') \
.config("spark.ui.port", "4060") \
.getOrCreate()
In [20]:
#spark = SparkSession.builder \
# .master('local[8]') \
# .appName('Yelp Integradora') \
# .getOrCreate()
In [7]:
sc = spark.sparkContext
In [8]:
spark#.stop()
Out[8]:
Importando as Bases Origem - Raw¶
- A princípio todas as bases foram importadas para análise e entendimento de oportunidades de análise.
In [9]:
usr_raw = spark.read.json(datapath+'/yelp_academic_dataset_user.json')
rv_raw = spark.read.json(datapath+'/yelp_academic_dataset_review.json')
bz_raw = spark.read.json(datapath+'/yelp_academic_dataset_business.json')
tp_raw = spark.read.json(datapath+'/yelp_academic_dataset_tip.json')
ch_raw = spark.read.json(datapath+'/yelp_academic_dataset_checkin.json')
In [10]:
# Visualizando Estrutura
#bz_raw.printSchema()
Tratamento das bases¶
Check-ins¶
- Explosão das datas na base de Check-ins para análise dos principais dias e horários de movimento nos restaurantes.
In [11]:
ch = ch_raw.withColumn('dt', f.explode(f.split(f.col('date'),', ')))
In [12]:
ch_raw.show()
In [13]:
ch.createOrReplaceTempView('ch')
- Transformação da data em unidades de tempo para análise
In [14]:
ch1 = spark.sql('''
SELECT business_id, dt,
DATE(dt) AS date,
DAYOFWEEK(dt) AS dow,
MONTH(dt) AS month,
YEAR(dt) AS year,
HOUR(dt) AS hour,
DAY(dt) AS day
FROM ch
''')
In [15]:
ch1.show()
In [16]:
ch1.createOrReplaceTempView('ch1')
In [17]:
ch2 = spark.sql('''
SELECT business_id, year, month, dow, day,
COUNT(date) AS count
FROM ch1
GROUP BY business_id, year, month, dow, day
ORDER BY count DESC
''')
In [18]:
ch2.createOrReplaceTempView('ch2')
Analisando o movimento de um estabelecimento¶
- Como exemplo, foi selecionado aleatóriamente um estabelecimento de Toronto. A seguir é analisada a quantidade de check-ins por mês e dia da semana.
In [19]:
df = spark.sql('''
SELECT *
FROM ch2
WHERE business_id = 'BxCzy1WOVxOrr_G7V4BIvg'
''').toPandas()
In [20]:
df2 = df.groupby(['month','dow'])['count'].sum().unstack('dow').fillna(0)
In [21]:
plt.figure(figsize = (25,7))
sns.heatmap(data = df2, cmap="YlGnBu", annot = True)
Out[21]:
Pelo gráfico, verifica-se que o local é mais movimentado entre janeiro e março, principalmente aos finais de semana (dias 7 e 1). Para a modelagem, o ideal seria utilizar a moda (valor mais frequente) ou mediana dos check-ins, mas não é possível calcular com processento paralelo. A média pode não ser representativa, pois o pico de movimento está concentrado nos finais de semana.
Tips¶
- No início do estudo consideramos a utilização das informações da base Tips.
In [23]:
tp_raw.createOrReplaceTempView('tp')
In [24]:
tp_raw.show()
- Quantidade de tips por estabelecimento e usuário. Considerando que temos duas bases com texto de REVIEWS e TIPS. Na base TIPS vamos explorar a possibilidade de criação de novas variáveis.
In [25]:
tp_usr = spark.sql('''
SELECT user_id,
count(text) AS tips_counter,
sum(compliment_count) as total_compliments
FROM tp
GROUP BY user_id
ORDER BY total_compliments DESC
''')
In [26]:
tp_usr.show()
In [27]:
tp_usr.createOrReplaceTempView('tp_usr')
In [28]:
tp_bz = spark.sql('''
SELECT business_id,
count(text) AS tips_counter,
sum(compliment_count) as total_compliments
FROM tp
GROUP BY business_id
ORDER BY total_compliments DESC
''')
In [29]:
tp_bz.createOrReplaceTempView('tp_bz')
In [30]:
tp_bz.show()
A quantidade total e o alcance das dicas por estabelecimento serão utilizadas na modelagem.
Business¶
Abertura dos atributos e horários para colunas¶
- Para incluir os atributos e dias da semana na tabela, é necessário extrair as listas das colunas 'attributes' e 'hours'.
In [32]:
dfs = []
for x in ["hours", "attributes"]:
cols = bz_raw.select(f"{x}.*").columns
for col in cols:
try:
dfs.append(dfs[-1].withColumn(col, f.col(f"{x}.{col}")))
except IndexError:
dfs.append(bz_raw.withColumn(col, f.col(f"{x}.{col}")))
bz = dfs[-1].drop("hours", "attributes")
bz.createOrReplaceTempView("bz")
bz.printSchema()
Limpeza e padronização dos atributos¶
- Dentro do campo de Atributos havia um vetor com uma série de informações que precisamos transofrmar em novas colunas para serem utilizadas na modelagem
- Foram filtrados APENAS estabelecimentos ABERTOS de TORONTO - Delimitação da análise exploratória.
In [33]:
bz1 = spark.sql('''
SELECT *
FROM bz
WHERE city == 'Toronto'
AND is_open == 1
''')
In [34]:
bz1.groupBy('is_open').count().show()
In [35]:
cols = bz_raw.select('attributes.*').columns
Valores únicos em cada coluna de atributos. As colunas que contém listas serão desconsideradas da base para facilitar a análise. As demais serão codificadas conforme a seguir:
Atributos com True/False como resposta:
- Null/None: 0
- False: 1
- True: 2
Atributos que possuem uma descrição das características foram substituídas por números como o exemplo:
- Null/None: 0
- Característica 1 - 1
- Característica N - N
In [36]:
[bz1.groupBy(cols[x]).count().show() for x in range(len(cols))]
Out[36]:
- Com base no entendimento das informações e distribuições acima, foram definidos os campos que permaneceriam na base e quais seriam removidos (
unwanted).
In [37]:
unwanted = ['Ambience','BestNights','BusinessAcceptsBitcoin','BusinessParking','BYOBCorkage','DietaryRestrictions','GoodForMeal','HairSpecializesIn','Music','Open24Hours','RestaurantsCounterService']
In [38]:
bz2 = bz1.drop('Ambience','BestNights','BusinessAcceptsBitcoin','BusinessParking','BYOBCorkage','DietaryRestrictions','GoodForMeal','HairSpecializesIn','Music','Open24Hours','RestaurantsCounterService',
'Sunday','Monday','Tuesday','Wednesday','Thursday','Friday','Saturday','address','postal_code','state','city','is_open')
In [39]:
for col in unwanted:
cols.remove(col)
In [40]:
bz3 = bz2.fillna('0')
Codificação dos atributos conforme descrito anteriormente. O processamento com REGEX foi mais eficiente do que a função .replace do PySpark.
In [42]:
bz4 = bz3.replace({'True':'2', 'False':'1','None':'0','null':'0'}, cols)
In [43]:
bz5 = bz4.withColumn('AgesAllowed',f.regexp_replace(f.col('AgesAllowed'), "u'allages'",'0'))
bz6 = bz5.withColumn('AgesAllowed',f.regexp_replace(f.col('AgesAllowed'), "u'19plus'",'19'))
In [44]:
bz7 = bz6.withColumn('Alcohol',f.regexp_replace(f.col('Alcohol'), "'none'|u'none'",'0'))
bz8 = bz7.withColumn('Alcohol',f.regexp_replace(f.col('Alcohol'), "'beer_and_wine'|u'beer_and_wine'",'1'))
bz9 = bz8.withColumn('Alcohol',f.regexp_replace(f.col('Alcohol'), "'full_bar'|u'full_bar'",'2'))
In [45]:
bz10 = bz9.withColumn('NoiseLevel',f.regexp_replace(f.col('NoiseLevel'), "'quiet'|u'quiet'",'1'))
bz11 = bz10.withColumn('NoiseLevel',f.regexp_replace(f.col('NoiseLevel'), "'average'|u'average'",'2'))
bz12 = bz11.withColumn('NoiseLevel',f.regexp_replace(f.col('NoiseLevel'), "'very_loud'|u'very_loud'",'4'))
bz13 = bz12.withColumn('NoiseLevel',f.regexp_replace(f.col('NoiseLevel'), "'loud'|u'loud'",'3'))
In [46]:
bz14 = bz13.withColumn('Smoking',f.regexp_replace(f.col('Smoking'), "'no'|u'no'",'1'))
bz15 = bz14.withColumn('Smoking',f.regexp_replace(f.col('Smoking'), "'outdoor'|u'outdoor'",'2'))
bz16 = bz15.withColumn('Smoking',f.regexp_replace(f.col('Smoking'), "'yes'|u'yes'",'3'))
In [47]:
bz17 = bz16.withColumn('WiFi',f.regexp_replace(f.col('WiFi'), "'no'|u'no'",'1'))
bz18 = bz17.withColumn('WiFi',f.regexp_replace(f.col('WiFi'), "'paid'|u'paid'",'2'))
bz19 = bz18.withColumn('WiFi',f.regexp_replace(f.col('WiFi'), "'free'|u'free'",'3'))
In [48]:
bz20 = bz19.withColumn('RestaurantsAttire',f.regexp_replace(f.col('WiFi'), "'casual'|u'casual'",'1'))
bz21 = bz20.withColumn('RestaurantsAttire',f.regexp_replace(f.col('WiFi'), "'dressy'|u'dressy'",'2'))
bz22 = bz21.withColumn('RestaurantsAttire',f.regexp_replace(f.col('WiFi'), "'formal'|u'formal'",'3'))
In [49]:
#[bz22.groupBy(cols[x]).count().show() for x in range(len(cols))]
In [50]:
bz22.createOrReplaceTempView('bz22')
Join da quantidade de tips por estabelecimento¶
- Inclusão de variáveis "Quantidade de Tips" e "Quantidade de Elogios" por estabelecimento da Base TIPs.
In [51]:
bz23 = spark.sql('''
SELECT A.*,
IFNULL(B.tips_counter,0) AS tips_counter_bz,
IFNULL(B.total_compliments,0) AS total_compliments_bz
FROM bz22 as A
LEFT JOIN tp_bz as B
ON A.business_id = B.business_id
''')
In [52]:
bz23.columns
Out[52]:
In [53]:
bz23.write \
.format('csv') \
.mode('overwrite') \
.option('sep', ',') \
.option('header', True) \
.save('output/yelp_bz.csv')
In [54]:
bz23.createOrReplaceTempView('bz')
Users¶
- Preparação da base para montar clusterização por perfil de usuários no R.
- Para fazer recomendações baseadas no perfil do usuário, é necessário agrupá-los de alguma forma. Foram testados os algoritmos de clusterização hierárquica e k-means, mas apenas o último trouxe resultados interpretáveis e satisfatórios.
Seleção e tratamento da variáveis¶
In [56]:
rv_raw.createOrReplaceTempView('rv')
- Contadores de variáveis: "Quantidade de Amigos", "Quantidade de Anos que Usuário foi classificado como Elite" e "Ano em que ingressou no Yelp"
In [57]:
# Transforma variáveis para serem utilizadas na Matriz de Distâncias - H Clust
# Arrays em Numéricas - "id" de friends para friends_count
# Arrays em Numéricas - "ano" de elite para elite_count
# Data em Numéricas do ano - "yelping_since" para "year_since"
from pyspark.sql.functions import col, count, explode, split, array
from pyspark.sql.types import ArrayType, IntegerType
usr_raw1 = usr_raw.withColumn('friends_count', f.size(f.split(f.col('friends'), ' ')))
usr_raw1 = usr_raw1.withColumn('elite_count', f.size(f.split(f.col('elite'), ',')))
usr_raw1 = usr_raw1.withColumn('year_since', f.year('yelping_since'))
#usr_raw1.show()
In [58]:
usr_raw1.createOrReplaceTempView('usr_raw1')
- Seleção variáveis que serão utilizadas na clusterização realizada no R Studio
In [59]:
usr_base = spark.sql('''
SELECT
A.user_id,
A.average_stars,
A.compliment_cool,
A.compliment_cute,
A.compliment_funny,
A.compliment_hot,
A.compliment_list,
A.compliment_more,
A.compliment_note,
A.compliment_photos,
A.compliment_plain,
A.compliment_profile,
A.compliment_writer,
A.cool,
A.elite_count,
A.fans,
A.friends_count,
A.funny,
A.review_count as review_count_usr,
A.useful,
A.year_since,
B.business_id
FROM usr_raw1 as A LEFT JOIN rv as B
ON A.user_id = B.user_id
''')
In [60]:
usr_base.createOrReplaceTempView('usr_base')
In [61]:
spark.sql('''
SELECT
COUNT(DISTINCT user_id)
FROM usr_base
''').show()
- Seleção de usuários com reviews de business IDs em Toronto, já selecionados anteriormente na base de estabelecimentos
In [62]:
usr_base2 = spark.sql('''
SELECT
A.business_id,
B.*
FROM bz as A
LEFT JOIN usr_base as B
ON A.business_id = B.business_id
''')
In [63]:
usr_base2.createOrReplaceTempView('usr_base2')
In [64]:
spark.sql('''
SELECT
COUNT(DISTINCT user_id)
FROM usr_base2
''').show()
In [65]:
# Identificar User_id Apenas de Usuários que fizeram Reviews em Estabelecimentos Abertos Em Toronto
usr_base3 = spark.sql('''
SELECT
DISTINCT user_id
FROM usr_base2
''')
In [66]:
usr_base3.createOrReplaceTempView('usr_base3')
In [67]:
# Seleciona Apenas de Usuários que fizeram Reviews em Estabelecimentos Abertos Em Toronto
usr_base4 = spark.sql('''
SELECT
A.user_id,
A.average_stars,
A.compliment_cool,
A.compliment_cute,
A.compliment_funny,
A.compliment_hot,
A.compliment_list,
A.compliment_more,
A.compliment_note,
A.compliment_photos,
A.compliment_plain,
A.compliment_profile,
A.compliment_writer,
A.cool,
A.elite_count,
A.fans,
A.friends_count,
A.funny,
A.review_count AS review_count_usr,
A.useful,
A.year_since
FROM usr_raw1 as A LEFT JOIN usr_base3 as B
ON A.user_id = B.user_id
WHERE B.user_id <> ''
''')
In [68]:
usr_base4.createOrReplaceTempView('usr_base4')
- Quantidade inicial de usuários na base:
In [69]:
spark.sql('''
SELECT
COUNT(DISTINCT user_id)
FROM usr_raw1
''').show()
- Quantidade final de usuários na base:
In [70]:
spark.sql('''
SELECT
COUNT(DISTINCT user_id)
FROM usr_base4
''').show()
Join de tips por usuário¶
In [71]:
usr = spark.sql('''
SELECT A.*,
IFNULL(B.tips_counter,0) AS tips_counter,
IFNULL(B.total_compliments,0) AS total_compliments
FROM usr_base4 as A
LEFT JOIN tp_usr as B
ON A.user_id = B.user_id
''')
In [72]:
usr.count()
Out[72]:
In [71]:
usr.write \
.format('csv') \
.mode('overwrite') \
.option('sep', ',') \
.option('header', True) \
.save('output/yelp_usr.csv')
In [73]:
usr.createOrReplaceTempView('usr')
In [74]:
usr.columns
Out[74]:
Análise das bases Business e Users no R¶
- Business: Análise de componentes principais para entender a variabilidade da nota média dos estabelecimentos em relação aos atributos.
- Users: Cluseterização K-means para entende o perfil dos usuários.
Base para treino do modelo¶
Consolidação dass bases de USUÁRIO + BUSINESS + REVIEWS (apenas a nota que será a variável resposta utilizada na classificação)
Após consolidação da base em Python, será feita a leitura da base e substituição da variável resposta para binária:
- Notas maiores ou iguais a 4 - boas (1)
- Notas menores do que 4 - ruim (0)
Além disso, serão mantidas apenas as variáveis numéricas para treinamento da rede neural.
In [77]:
usr_clusters = spark.read.csv('output/usr_cluster.csv', header = True)
usr_clusters.createOrReplaceTempView('usr_clusters')
In [78]:
#Juntando número do cluster na base de usuários
usr_clusters2 = spark.sql("""
SELECT A.*,
B.cluster_usr
FROM usr as A
LEFT JOIN usr_clusters as B
ON A.user_id = B.user_id
""")
In [79]:
usr_clusters2.createOrReplaceTempView('usr_clusters2')
In [80]:
#seleção das reviews daqueles usuários
bf1 = spark.sql("""
SELECT
B.*,
A.business_id,
A.stars as stars_rv,
YEAR(A.date) AS year_rv
FROM usr_clusters2 as B
LEFT JOIN rv as A
ON B.user_id = A.user_id
""")
In [81]:
bf1.columns
Out[81]:
In [82]:
bf1.createOrReplaceTempView('bf1')
- Para evitar informações muito desatualizadas na base, serão consideradas apenas reviews a partir de 2017, que corresponde à maior parte da base de reviews, como é possível ver no gráfico abaixo.
In [83]:
rv_df = spark.sql("""
SELECT stars_rv, year_rv, count(stars_rv) as qtde_rv
FROM bf1 as A
GROUP BY stars_rv, year_rv
""").toPandas()
In [84]:
rv_df.groupby(['year_rv','stars_rv'])['qtde_rv'].agg('sum').unstack('stars_rv').plot(kind = 'bar', stacked = True, figsize = (25,6))
Out[84]:
In [85]:
bf2 = spark.sql('''
SELECT A.*,
B.categories,
B.latitude,
B.longitude,
B.name,
B.review_count,
B.stars,
B.AcceptsInsurance,
B.AgesAllowed,
B.Alcohol,
B.BYOB,
B.BikeParking,
B.BusinessAcceptsCreditCards,
B.ByAppointmentOnly,
B.Caters,
B.CoatCheck,
B.Corkage,
B.DogsAllowed,
B.DriveThru,
B.GoodForDancing,
B.GoodForKids,
B.HappyHour,
B.HasTV,
B.NoiseLevel,
B.OutdoorSeating,
B.RestaurantsAttire,
B.RestaurantsDelivery,
B.RestaurantsGoodForGroups,
B.RestaurantsPriceRange2,
B.RestaurantsReservations,
B.RestaurantsTableService,
B.RestaurantsTakeOut,
B.Smoking,
B.WheelchairAccessible,
B.WiFi,
B.tips_counter_bz,
B.total_compliments_bz
FROM bf1 as A
RIGHT JOIN bz as B
ON A.business_id = B.business_id
WHERE year_rv >= 2017
''')
In [86]:
bf2.createOrReplaceTempView('bf2')
In [87]:
spark.sql('''
SELECT
COUNT(DISTINCT business_id)
FROM bf2
''').show()
- A seguir está a estrutura de dados de entrada no Modelo:
In [88]:
bf2.columns
Out[88]:
In [89]:
bf2.write \
.format('csv') \
.mode('overwrite') \
.option('sep', ',') \
.option('header', True) \
.save('output/yelp.csv')
Reviews¶
- Como próximos passos, seria interessante entender melhor as características de cada cluster.
- Para entender melhor a natureza de cada cluster de usuários definida no R, utilizamos a nuvem de palavras para encontrar similaridade de perfil.
Tratamento do texto¶
In [90]:
#Função para remoção de pontuação e tratamento das palavras
def word_clean(sdf,col,new_col):
rv1 = sdf.withColumn(new_col,f.regexp_replace(f.col(col), "'d", " would"))
rv2 = rv1.withColumn(new_col,f.regexp_replace(f.col(new_col), "'ve", " have"))
rv3 = rv2.withColumn(new_col,f.regexp_replace(f.col(new_col), "'s", " is"))
rv4 = rv3.withColumn(new_col,f.regexp_replace(f.col(new_col), "'re", " are"))
rv5 = rv4.withColumn(new_col,f.regexp_replace(f.col(new_col), "n't", " not"))
rv6 = rv5.withColumn(new_col,f.regexp_replace(f.col(new_col), '\W+', " "))
rv7 = rv6.withColumn(new_col,f.lower(f.col(new_col)))
return rv7
In [91]:
rv_raw.columns
Out[91]:
In [92]:
usr_clusters2.createOrReplaceTempView('usr_clusters2')
rv_raw.createOrReplaceTempView('rv_raw')
In [93]:
spark.sql('''
SELECT
count(*)
FROM rv_raw
''').show()
In [94]:
words = spark.sql('''
SELECT
A.review_id,
A.business_id,
A.user_id,
A.text,
max(B.text) as text_tp
FROM rv_raw as A
LEFT JOIN tp as B
ON A.user_id = B.user_id and A.business_id = B.business_id
GROUP BY 1, 2, 3, 4
''')
In [95]:
words.createOrReplaceTempView('words')
In [96]:
spark.sql('''
SELECT
count(*)
FROM words
''').show()
In [97]:
bz.columns
Out[97]:
In [98]:
words2 = spark.sql('''
SELECT A.*
FROM bz22 as B
LEFT JOIN words as A
ON A.business_id = B.business_id
''')
In [99]:
words2.count()
Out[99]:
In [100]:
words2.createOrReplaceTempView('words2')
In [101]:
words3 = spark.sql('''
SELECT A.*,
B.cluster_usr
FROM words2 as A
LEFT JOIN usr_clusters2 as B
ON A.user_id = B.user_id
''')
In [102]:
words3.count()
Out[102]:
- Chama a função para limpar os textos definida anteriormente
In [103]:
words4 = word_clean(words3,'text','text_clean')
words5 = word_clean(words4,'text_tp','text_clean_tp')
In [104]:
words5.show(4)
In [105]:
from pyspark.sql.functions import split, regexp_replace, col, split, stddev, regexp_replace
words6 = words5.select("cluster_usr", split(
"text_clean", r"[\s.]+").alias("text2"), split("text_clean_tp", r"[\s.]+").alias("text_tp2"))
- Utiliza a função nativa do PySpark
STOPWORDS REMOVERpara remoção de "palavras de parada" que não contribuirão na análise das palavras por não transmitirem significados para os clusters.
In [106]:
#Configuração de idioma do Java que afeta o pacote StopWords
locale = sc._jvm.java.util.Locale
locale.setDefault(locale.forLanguageTag("en-US"))
In [107]:
remover = StopWordsRemover(inputCol="text2", outputCol="text_filtered")
words7 = remover.transform(words6)
words7.show(5)
In [108]:
from pyspark.sql.functions import concat_ws
words8 = words7.select("cluster_usr", "text_filtered",
concat_ws(' ', "text_filtered").alias("text_filtered2"))
In [109]:
words9 = words8.withColumn('wordCount', f.size(f.split(f.col('text_filtered2'),' ')))
In [110]:
words10 = words9.withColumn('word', f.explode(f.split(f.col('text_filtered2'), ' ')))\
.groupBy('word','cluster_usr')\
.count()\
.sort('count', ascending = False)
words10.show()
In [115]:
words10.createOrReplaceTempView('words10')
In [116]:
cls = []
#for i in tqdm(range(1, 11)): # Biblioteca que apresenta a evolução do tempo de processamento, mas não está instalada no servidor AWS
for i in range(1, 11):
cls.append(spark.sql(f'''
SELECT word,
count
FROM words10
WHERE word IS NOT NULL
AND cluster_usr == {i}
ORDER BY count DESC limit 50
'''))
cls[i-1] = [tuple(x) for x in cls[i-1].toPandas().to_numpy()]
In [119]:
cluster = 1
for cl in cls:
print(f"Cluster {cluster}")
plt.figure(figsize=(10, 10))
plt.imshow(WordCloud().generate(str(cl).replace("'", "")), interpolation='bilinear')
plt.axis('off')
plt.show()
cluster += 1
- Observamos que a maior parte dos cluster estão relacionados ao setor de alimentos e restaurantes com palavras em comum como "Food", "good", "chicken", "delicious", ...
- No cluster 6, observamos uma natureza relativamente diferente com aparecimento de palavras como "hotel", "holiday", "store", "conference" dando a impressão de ser um cluster de turistas.
Heatmap¶
- Criando mapa de calor da concentração de reviews
In [120]:
base_mapas = spark.sql('''
SELECT A.*, B.latitude, B.longitude
FROM bz22 as B
LEFT JOIN words as A
ON A.business_id = B.business_id
''')
In [121]:
base_mapas.columns
Out[121]:
In [122]:
base_mapas.createOrReplaceTempView('base_mapas')
In [123]:
mapa1 = spark.sql("""
SELECT latitude,
longitude
FROM base_mapas
WHERE latitude is not null
AND longitude is not null
""")
Decobrindo o ponto central de Latitude e Longetude do Mapa
In [124]:
spark.sql("""
SELECT avg(latitude) as avg_lat,
avg(longitude) as avg_long
FROM base_mapas
""").show()
In [125]:
import folium
from folium import plugins
mapa = folium.Map(location=[43.673, -79.391],
zoom_start=11,
tiles='Stamen Toner')
# OpenStreetMap, Stamen Terrain, Stamen Toner
In [126]:
mapa
lat = mapa1.toPandas()['latitude'].values
lon = mapa1.toPandas()['longitude'].values
In [127]:
coordenadas = []
for la, lo in zip(lat, lon):
coordenadas.append([la,lo])
mapa.add_child(plugins.HeatMap(coordenadas))
Out[127]: